
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.Path;

import com.ccxe.parser.Filter;
import com.ccxe.parser.InjectHour;
import com.ccxe.parser.InjectMin;
import com.ccxe.parser.Selector;
import com.ccxe.util.DateConvertUtil;

public class Parser {

	private static final Log LOG = LogFactory.getLog(Parser.class);

	public static void main(String[] args) throws Exception {
		//获取传入参数
		if (args.length < 1) {
			System.err.println("Usage: Parser <input1>[,input2,input3....] -db <database_name> ");
			return;
		}
		//解析参数
		Path db = new Path("db_" + DateConvertUtil.getDate());
		for (int i = 1; i < args.length; i++) {
			if ("-db".equals(args[i])) {
				db = new Path(args[i + 1]);
				i++;
			}
		}
		LOG.info("Parser started   " + DateConvertUtil.getDate());
		if (LOG.isInfoEnabled()) {
			LOG.info("Database: " + db);
			LOG.info("Input:" + args[0]);
		}

		String[] pathString = args[0].split(",");
		Path [] inputs = new Path [pathString.length];
		for (int i = 0; i < pathString.length; i++) {
			inputs[i] = new Path(pathString[i]);
		}
		//生成文件目录
		Path select_path = new Path(db + "/select");
		Path injectHour_path = new Path(db+"/inject_hour");
		Path insertMin_path = new Path(db + "/insert_min");
		Path filter_path = new Path(db+"/filter");
		//相应的子操作
		Selector.selcet(inputs, select_path);
		InjectHour.insert(select_path, injectHour_path);
		InjectMin.insert(injectHour_path, insertMin_path);
		Filter.insert(insertMin_path, filter_path);
		
		if (LOG.isInfoEnabled()) {
			LOG.info("Parser finished: " + db);
		}
	}
}
